package com.baomidou.kisso.service;

import java.util.concurrent.Executors;

import org.springframework.stereotype.Service;

import cn.hutool.core.thread.ThreadUtil;
import io.reactivex.rxjava3.core.Scheduler;
import io.reactivex.rxjava3.core.Single;
import io.reactivex.rxjava3.schedulers.Schedulers;
import lombok.extern.slf4j.Slf4j;

@Service
@Slf4j
public class RxjavaService {

    // 创建线程池
    public static final Scheduler IO = Schedulers
            .from(Executors.newFixedThreadPool(10, new NamedThreadFactory("A001")));
    public static final Scheduler IO1 = Schedulers
            .from(Executors.newFixedThreadPool(10, new NamedThreadFactory("B101")));

    public void downAndUploadFile(String name) {
        // 使用rxjava 先从外网下载文件，再把文件上传到内网
        Single.create((emiter) -> {
            log.info("dowm  ：{} , 当前线程 ：{} ", name, Thread.currentThread().getName());
            ThreadUtil.sleep(5000);
            emiter.onSuccess("下载成功");
        }).subscribeOn(IO)
                .map((result) -> {
                    log.info("map  ：{} , 当前线程 ：{} ", result, Thread.currentThread().getName());
                    return result + " map转换";
                }).onErrorResumeNext((ex) -> {
                    log.error("map处理异常 :{} ", ex);
                    return Single.error(ex);
                }).observeOn(IO1)
                .subscribe((result) -> {
                    log.info("upload  ：{} , 当前线程 ：{} ", result, Thread.currentThread().getName());
                    ThreadUtil.sleep(5000);
                    log.info("upload  end ");
                }, (error) -> {
                    log.error("error  ：{} ", error.getMessage());
                });

    }

}
